package com.lilei.bigdata.mr.friend;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

/**
 * 找出来每一个好友对应的用户  然后将用户组合在一起 就可以找到共同好友
 *
 * A	I,K,C,B,G,F,H,O,D,
 * B	A,F,J,E,
 * C	A,E,B,H,F,G,K,
 * D	G,C,K,A,L,F,E,H,
 * E	G,M,L,H,A,F,B,D,
 * F	L,M,D,C,G,A,
 * G	M,
 * H	O,
 * I	O,C,
 * J	O,
 * K	B,
 * L	D,E,
 * M	E,F,
 * O	A,H,I,J,F,
 */
public class ShareFriendStepOne {
    static class SharedFriendsStepOneMapper extends Mapper<LongWritable,Text,Text,Text>{
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String[] person_friends = line.split(":");
            String person = person_friends[0];
            String friends = person_friends[1];
            for (String friend:friends.split(",")){
                context.write(new Text(friend),new Text(person));
            }
        }
    }
    static class SharedFriendsStepOneReducer extends Reducer<Text,Text,Text,Text>{
        @Override
        protected void reduce(Text friend, Iterable<Text> persons, Context context) throws IOException, InterruptedException {
            StringBuffer stringBuffer =new StringBuffer();
            for (Text person:persons){
                stringBuffer.append(person).append(",");
            }

            context.write(friend,new Text(stringBuffer.toString()));
        }
    }
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();

        Job job = Job.getInstance(conf);
        job.setJarByClass(ShareFriendStepOne.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        job.setMapperClass(SharedFriendsStepOneMapper.class);
        job.setReducerClass(SharedFriendsStepOneReducer.class);

        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        job.waitForCompletion(true);
    }
}
